package com.danan.realtime.app.dws;

import com.danan.realtime.app.BaseSqlApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.function.SplitWord;
import com.danan.realtime.pojo.KeywordBean;
import com.danan.realtime.util.SinkUtil;
import com.danan.realtime.util.SourceUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import scala.collection.immutable.Stream;

/**
 * @author NanHuang
 * @Date 2023/1/26
 */
public class DwsApp_1_traffic_keyword_page_view_window extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsApp_1_traffic_keyword_page_view_window().init(5001,3,"DwsApp_1_traffic_keyword_page_view_window");
    }
    @Override
    public void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 建表,与dwd_traffic_page中的 数据关联
        readDwdTrafficPage(tableEnv);
        // 2 过滤出搜索记录
        filterKeyword(tableEnv);
        // 3 关键词分词
        splitKeyword(tableEnv);
        // 4 开窗聚合，每5秒统计关键词的搜索次数(TVF)
        Table result = windowAndAgg(tableEnv);
        // 5 将汇总的结果写出到Clickhouse中
        writeToClickhouse(result,env,tableEnv);
    }

    private void writeToClickhouse(Table result,StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 将table转为流
        DataStream<Tuple2<Boolean, KeywordBean>> ds = tableEnv.toRetractStream(result, KeywordBean.class);
        // 2 过滤撤回流中的旧数据
        SingleOutputStreamOperator<KeywordBean> kwDs = ds.filter(t -> t.f0).map(t -> t.f1);
        // 3 将结果写入Clickhouse中（JDBC Sink）
        kwDs.addSink(SinkUtil.getClickhouseSink(Constant.TOPIC_DWS_TRAFFIC_KEYWORD_PAGE_VIEW_WINDOW,KeywordBean.class));
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private Table windowAndAgg(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select " +
                "   date_format(window_start,'yyyy-MM-dd HH:mm:ss') stt," +
                "   date_format(window_end,'yyyy-MM-dd HH:mm:ss') edt," +
                "   keyword," +
                "   count(1) keywordCount," +
                "   unix_timestamp() ts " +
                "from table (" +
                "   tumble(table split_result,descriptor(et),interval '5' seconds)" +
                ")" +
                "group by window_start,window_end,keyword");
    }

    private void splitKeyword(StreamTableEnvironment tableEnv) {
        // 注册自定义分词函数
        tableEnv.createTemporaryFunction("split_keyword",new SplitWord());
        // 分词
        Table splitResult = tableEnv.sqlQuery("select " +
                "keyword," +
                "et " +
                "from keyword " +
                "join lateral table (split_keyword(item)) on true");
        tableEnv.createTemporaryView("split_result",splitResult);
    }

    private void filterKeyword(StreamTableEnvironment tableEnv) {
        Table keyword = tableEnv.sqlQuery("select " +
                "item," +
                "et " +
                "from traffic_page " +
                "where item is not null " +
                "and item_type='keyword' " +
                "and last_page_id in ('keyword','search')");
        tableEnv.createTemporaryView("keyword",keyword);
    }

    private void readDwdTrafficPage(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table traffic_page (" +
                "item string," +
                "item_type string," +
                "last_page_id string," +
                "ts bigint," +
                "et as to_timestamp_ltz(ts,3)," +
                "watermark for et as et - interval '3' seconds " +
                ")" +
                SourceUtil.getKafkaWith(Constant.TOPIC_DWD_TRAFFIC_PAGE,"DwsApp_1_traffic_keyword_page_view_window"));
    }
}
